package com.mall.hadoop.job.diffpeople;

import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.mall.common.dao.MysqlDBManager;
import com.mysql.jdbc.Connection;

/**
 * 全国人数类型统计reduce
 * @ClassName: DiffPeopleReduce
 * @Description: 
 * @author:wangwenyue wangwenyue@ide365.com
 * @date 2016年7月15日 下午2:51:41
 */
public class DiffPeopleReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
	
	private static List<Map<String, String>> allDistricts=new ArrayList<Map<String, String>>();
	
	static{
		allDistricts = getAllDistricts();
	}
	
	
	@Override
	protected void reduce(Text key, Iterable<IntWritable> values, Context context)
			throws IOException, InterruptedException {
		int count=0;
		for (IntWritable v : values) {
			 count += v.get();
		}
		System.out.println("=====key:"+key+"===="+count+"======》》地区一共有多少个："+allDistricts.size());
		
		String keyString = key.toString();
		
		
		String[] splitBaby = keyString.split("b"); //切割baby数据
		
		/**
		 * 裁剪baby数据  裁剪后长度小于1的都为垃圾数据均无视， 已baby开头的是计算的全国总数暂时不用。
		 */
		if(splitBaby.length>1 && !"baby".equals(keyString)){
			//标示省市地县的地区名称
			String signName = splitBaby[1]; 
			
//			如果是市
			if(signName.contains("1")){
				String[] signArea = signName.split("1");//切割市名称
				reckonColumnCount(signArea,1, "student_sum", count);
			}
			
//			如果是县区
			if(signName.contains("2")){
				String[] signCity = signName.split("2");//切割区县
				reckonColumnCount(signCity,2, "student_sum", count);
			}
			
//			如果是省
			if(signName.contains("3")){
				String[] signProvince = signName.split("3");//切割省名称
				reckonColumnCount(signProvince,3, "student_sum", count);
			}
		}
		
		
		
		
		/**
		 * 切割班级数据 统计老师
		 */
		String[] splitClass = keyString.split("c"); 
		if(splitClass.length>1 && !"class".equals(keyString) && !"c".equals(keyString)){
			//标示省市地县的地区名称
			String signName = splitClass[1]; 
			
//			如果是市区
			if(signName.contains("1")){
				String[] signArea = signName.split("1");//切割县名称
				reckonColumnCount(signArea,1, "teacher_sum", count);
			}
			
//			如果是县
			if(signName.contains("2")){
				String[] signCity = signName.split("2");//切割市区名称
				reckonColumnCount(signCity,2, "teacher_sum", count);
			}
			
//			如果是省
			if(signName.contains("3")){
				String[] signProvince = signName.split("3");//切割省名称
				reckonColumnCount(signProvince,3, "teacher_sum", count);
			}
		}
		
		
		
		/**
		 * 切割学校数据 显示园长数据
		 */
		String[] splitPrin = keyString.split("s");
		if(splitPrin.length>1 && !"class".equals(splitPrin) && !"s".equals(splitPrin)){
			String signName = splitPrin[1];  //地区名称
			System.out.println(signName);
			
//			如果是市区
			if(signName.contains("1")){
				String[] signArea = signName.split("1");//切割县名称
				reckonColumnCount(signArea,1, "principal_sum", count);
			}
			
//			如果是县
			if(signName.contains("2")){
				String[] signCity = signName.split("2");//切割市区名称
				reckonColumnCount(signCity,2, "principal_sum", count);
			}
			
//			如果是省
			if(signName.contains("3")){
				String[] signProvince = signName.split("3");//切割省名称
				reckonColumnCount(signProvince,3, "principal_sum", count);
			}
		}
		
		
		context.write(key, new  IntWritable(count));
	}
	
	
	/**
	 * 统计地区某项的总数
	 * @param splitArr 未格式化的地区
	 * @param column 统计的列名
	 * @param count 统计的总数
	 */
	public void reckonColumnCount(String[] splitArr,int type,String column,int count){
		if(splitArr.length==2){//如果正常数据的话
			String name = splitArr[1]; //取出地区名称
			String area_name="";
			String area_code="";
			//循环公共库
			for (Map<String, String> mps : allDistricts) {
				
				//市
				if(type==2){
					if(mps.get("aname").contains(name)){
						area_name = mps.get("aname"); //取出公共库的地区名称
						area_code = mps.get("acode");// 取出公共库的地区编码
						String studentSum = String.valueOf(count); //取到reduce计算的此地区的总数
						addDB(area_name, area_code, studentSum,column);//插入统计数据库
						break;
					}
				}
				
				//县区
				if(type==1){
					if(mps.get("cname").contains(name)){
						area_name = mps.get("cname"); //取出公共库的地区名称
						area_code = mps.get("ccode");// 取出公共库的地区编码
						String studentSum = String.valueOf(count); //取到reduce计算的此地区的总数
						addDB(area_name, area_code, studentSum, column);//插入统计数据库
						break;
					}
				}
				
				//省
				if(type==3){
					if(mps.get("pname").contains(name)){
						area_name = mps.get("pname"); //取出公共库的地区名称
						area_code = mps.get("pcode");// 取出公共库的地区编码
						String studentSum = String.valueOf(count); //取到reduce计算的此地区的总数
						addDB(area_name, area_code, studentSum,column);//插入统计数据库
						break;
					}
				}
				
				
				/*switch (type) {
					case 1://县
						if(mps.get("aname").contains(name)){
							area_name = mps.get("aname"); //取出公共库的地区名称
							area_code = mps.get("acode");// 取出公共库的地区编码
						}
						break;
					case 2:	//市区
						if(mps.get("cname").contains(name)){
							area_name = mps.get("cname"); //取出公共库的地区名称
							area_code = mps.get("ccode");// 取出公共库的地区编码
						}
						break;
					case 3: //省
						if(mps.get("pname").contains(name)){
							area_name = mps.get("pname"); //取出公共库的地区名称
							area_code = mps.get("pcode");// 取出公共库的地区编码
						}
						break;
					default:
						break;
				}*/
				
				
			}
		}
	}
	
	
	
	/**
	 * 保存到统计数据库
	 * @param area_name地区名称
	 * @param area_code地区编码
	 * @param sum统计的总数
	 * @param clum统计的哪一列
	 */
	public void addDB(String area_name,String area_code,String sum,String clum){
		//如果有值就update
		String sql ="";
		if(!updateDb(area_code,area_name)){
			 sql="insert into t_different_people(area_name,area_code,"+clum+",create_time,update_time) values('"+area_name+"','"+area_code+"','"+sum+"',now(),now())";
		}else{ //没有就insert
			 sql="update t_different_people set "+clum+"='"+sum+"',update_time=now() where area_code='"+area_code+"'";
		}
		Connection conn = MysqlDBManager.getConnection();
		PreparedStatement prep =null ;
		try {
			prep = conn.prepareStatement(sql);
			int executeUpdate = prep.executeUpdate();
			conn.commit();
			System.out.println(executeUpdate);
		} catch (SQLException e) {
			e.printStackTrace();
		}finally{
			try {
				prep.close();
				conn.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
	}
	
	
	
	/**
	 * 查询有没有值
	 * @param code
	 * @param name
	 * @return
	 */
	public boolean updateDb(String code,String name){
		//添加前 先查询有没有该省（市区）名称和code，如果没有就insert  
		String sql = "select * from t_different_people where area_code='"+code+"' and area_name='"+name+"'";
		Connection conn = MysqlDBManager.getConnection();
		PreparedStatement prep =null;
		ResultSet rs =null;
		try {
			prep = conn.prepareStatement(sql);
			rs= prep.executeQuery();
			return rs.next();
		} catch (SQLException e) {
			e.printStackTrace();
		}finally{
			try {
				conn.close();
				prep.close();
				rs.close();
			} catch (SQLException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return false;
	}
	
	
	
	/**
	 * 获取所有地区
	 * @return
	 */
	public static List<Map<String, String>> getAllDistricts(){
		//查询  编号 省 编号 市 编号 区 （pcode，pname，ccode，cname，acode，aname）
		String sql = "select p.code pcode,p.name pname,c.code ccode,c.name cname,a.code acode,a.name aname from db_archive.area a,db_archive.city c,db_archive.province p where p.code=c.provincecode and c.code=a.citycode"; 
		
		Connection conn = MysqlDBManager.getConnection();
		PreparedStatement prep=null;
		ResultSet rs=null;
		
		List<Map<String, String>> listRes= new ArrayList<Map<String, String>>();
		Map<String, String> mps =null;
		
		try {
			prep = conn.prepareStatement(sql);
			rs= prep.executeQuery();
			int col = rs.getMetaData().getColumnCount();
			while (rs.next()) {
				for (int i = 1; i <= col; i++) {
					mps= new HashMap<String, String>();
					mps.put("pcode", rs.getString("pcode"));
					mps.put("pname", rs.getString("pname"));
					mps.put("ccode", rs.getString("ccode"));
					mps.put("cname", rs.getString("cname"));
					mps.put("acode", rs.getString("acode"));
					mps.put("aname", rs.getString("aname"));
					listRes.add(mps);
				}
			}
			
			
		} catch (SQLException e) {
			e.printStackTrace();
		}finally{
			try {
				conn.close();
				prep.close();
				rs.close();
			} catch (SQLException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return listRes;
	}
}
